package com.fc.push.handler.pub.service;

import java.io.UnsupportedEncodingException;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.fc.commons.security.MD5Util;
import com.fc.push.ServerConfig;
import com.fc.push.exception.ExcetionLevelConst;
import com.fc.push.exception.MServerException;
import com.fc.push.handler.pub.exec.PublishTaskPool;
import com.fc.push.handler.pub.exec.task.MqttPublishTask;
import com.fc.push.mqtt.SharedMqttClientPoolFacade;

@Component
public class MqttPublishUtil {

	private static final String DEFAULT_ENCODE = "UTF-8";

	private static int SLICE_SIZE = 1000;

	public MqttPublishUtil() {}
	
	

	/**
	 * 直接通过订阅主题来推送消息
	 * 
	 * @param message
	 * @param tags
	 */
	public boolean sendByTag(MqttMessage message, Object... tags) {
		int retryTimes = 3;
		MqttClient client = SharedMqttClientPoolFacade.getMqttClient();
		if (client.isConnected()) {
			for (Object tag : tags) {
				try {
					client.publish(tag.toString(), message);
				} catch (MqttPersistenceException e) {
					throw new MServerException(e, e.getMessage(), ExcetionLevelConst.EXCEPTION_LEVEL_2);
				} catch (MqttException e) {
					throw new MServerException(e, e.getMessage(), ExcetionLevelConst.EXCEPTION_LEVEL_2);
				}
			}
			return true;
		} else {
			while (--retryTimes > 0) {
				client = SharedMqttClientPoolFacade.getMqttClient();
				sendByTag(message, tags);
			}
		}
		return false;
	}
	
	
	private static final int DEFAULT_SLICE_QOS_LEVEL = 0;
	private static final boolean DEFAULT_SLICE_QOS_RETAINED = false;
	/**
	 * 通过分片publish消息到mosquitto
	 * 
	 * @param appkey
	 *            appkey
	 * @param payload
	 *            推送消息体
	 * @param objs
	 *            推送消息接收者列表
	 */
	public void pubMessageBySlice(String appkey, MqttMessage payload, Object... objs) {
		int totalSize = objs.length;
		int slice = totalSize / (SLICE_SIZE + 1) + 1;
		//通过分片来发时，QOS设置为0，RETAINED设置为false，只发送一次
		payload.setQos(DEFAULT_SLICE_QOS_LEVEL);
		payload.setRetained(DEFAULT_SLICE_QOS_RETAINED);
		/**
		 * 最大分成处理器核数+1片
		 */
		if (ServerConfig.getMaxThreadSize() * SLICE_SIZE < totalSize) {
			SLICE_SIZE = totalSize / ServerConfig.getMaxThreadSize() + 1;
		}

		for (int i = 0; i < slice; i++) {
			// List<String> uuids = new ArrayList<>();
			int begainIndex = i * SLICE_SIZE;
			int endIndex = begainIndex + SLICE_SIZE;
			if (endIndex > totalSize) {
				int size = totalSize - begainIndex;
				Object[] destObjs = new Object[size];
				System.arraycopy(objs, begainIndex, destObjs, 0, size);
				PublishTaskPool.Instance.getInstance().execute(new MqttPublishTask(appkey, payload, destObjs));
				break;
			}
			int size = endIndex - begainIndex;
			Object[] destObjs = new Object[size];
			System.arraycopy(objs, begainIndex, destObjs, 0, size);
			PublishTaskPool.Instance.getInstance().submit(new MqttPublishTask(appkey, payload, destObjs));
		}
	}

	private static final int DEFAULT_QOS_LEVEL = 1;
	private static final boolean DEFAULT_RETAINED = true;

	/**
	 * 生成客户端推送id
	 * 
	 * @param appkey
	 * @param deviceid
	 * @return
	 */
	public static String assClientId(String appkey, String deviceid) {
		if (appkey == null || deviceid == null) {
			throw new MServerException("Illegal argument");
		}
		return MD5Util.Instance.getInstance().md5TO23Bit(appkey + deviceid);
	}

	/**
	 * 生成Android推送包
	 * 
	 * @param message
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public MqttMessage assMqttPayload(JSONObject message) throws UnsupportedEncodingException {
		MqttMessage payload = null;
		payload = new MqttMessage(message.toJSONString().getBytes(DEFAULT_ENCODE));
		payload.setQos(DEFAULT_QOS_LEVEL);
		payload.setRetained(DEFAULT_RETAINED);
		return payload;
	}
}
